package com.cl.ks.flow.handler;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.ks.entity.KsFilterStatistics;
import com.cl.ks.service.KsFilterStatisticsService;
import com.cl.spark.dto.SparkResult;
import com.cl.ks.entity.ClueAutoResult;
import com.cl.ks.entity.KsRule;
import com.cl.ks.flow.KsFlowGraphHandler;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.flow.util.RuleUtil;
import com.cl.ks.flow.util.StaticCollection;
import com.cl.ks.service.ClueAutoResultService;
import com.querydsl.jpa.impl.JPAQueryFactory;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;

import static com.cl.ks.constant.QTableConstant.qKsFilterStatistics;
import static com.cl.ks.constant.QTableConstant.qKsRule;

@Component
public class KsRuleHandler extends BaseFlowNodeHandler {

    @Autowired
    private ClueAutoResultService clueAutoResultService;
    @Autowired
    private JPAQueryFactory queryFactory;
    @Autowired
    private RuleUtil ruleUtil;
    @Autowired
    private KsFilterStatisticsService ksFilterStatisticsService;

    @Override
    public SparkResult process(BaseFlowNodeHandler.ProcessParam processParam) {
        JSONObject expression = processParam.getSparkNode().getNodeExpression();
        List<String> dataTypeCodeList = (List<String>) expression.get("dataTypeCodes");
        if (dataTypeCodeList == null) {
            dataTypeCodeList = new ArrayList<>();
        }
        processParam.getKsFlowGraph().getKsDataTypeCodeList().addAll(dataTypeCodeList);

        Boolean isInvoke = expression.getBoolean("isInvoke");

        KsRule ksRule = new KsRule();
        String ruleName = expression.getString("ruleName");
        String ruleId = expression.getString("id");
        String ruleExpression = expression.getString("ruleExpressionJsonStr");
        if (ruleExpression == null) {
            ruleExpression = expression.getString("ruleExpressionJson");

        }
        ksRule.setRuleName(ruleName);
        ksRule.setId(Integer.valueOf(ruleId));
        ksRule.setRuleExpressionJson(ruleExpression);
        if (isInvoke) {

            KsRule invoke = queryFactory.selectFrom(qKsRule).where(qKsRule.id.eq(Integer.valueOf(ruleId))).fetchFirst();
            if (invoke != null) {
                ksRule = invoke;
            }
        }

        processParam.getKsFlowGraph().getRuleIdList().add(String.valueOf(ksRule.getId()));

        SparkResult originalSparkResult = processParam.getSparkNode().getInputSparkResultList().values().stream().findFirst().get();

        SparkResult ruleResult = ruleUtil.handleRule(ksRule, originalSparkResult);
        Dataset<Row> ruleResultDataset = ruleResult.getDataset();
        Long ruleResultTotal = ruleResult.getCount();

        processParam.getKsFlowGraph().setRuleDataTotal((int) (processParam.getKsFlowGraph().getRuleDataTotal() + ruleResultTotal));
        Integer pushClueAutoResultFlag = (Integer) expression.get("pushClueAutoResultFlag");
        if (pushClueAutoResultFlag != null && pushClueAutoResultFlag == 1) {

            String filterName =  processParam.getKsFlowGraph().getName();

            Integer ksFlowGraphId = processParam.getKsFlowGraph().getId();
            if (processParam.getKsFlowGraph().getIncrementFlag() == null || !processParam.getKsFlowGraph().getIncrementFlag()) {
                ClueAutoResult param = new ClueAutoResult();
                param.setRuleId(ksRule.getId());
                param.setFilterId(ksFlowGraphId);
                param.setFilterName(filterName);
                clueAutoResultService.deleteAll(param);
            }

            StaticCollection.newClueAutoResultList(ksFlowGraphId);

            String power = processParam.getKsFlowGraph().getResultPower() + "," + processParam.getKsFlowGraph().getPower();
            List<String> fieldList = Arrays.asList(ruleResultDataset.schema().fieldNames());
            if (!fieldList.contains("id")) {
                Optional<String> idField = fieldList.stream().filter(field -> field.contains("id")).findFirst();
                if (idField.isPresent()) {
                    ruleResultDataset = ruleResultDataset.withColumnRenamed(idField.get(), "id");
                } else {
                    ruleResultDataset = ruleResultDataset.withColumn("id", functions.lit(""));
                }
            }
            Integer finalRuleId = ksRule.getId();
            String finalRuleName = ksRule.getRuleName();
            ruleResultDataset.foreach(item -> {

                ClueAutoResult clueAutoResult = new ClueAutoResult();
                clueAutoResult.setRuleId(finalRuleId);
                clueAutoResult.setRuleName(finalRuleName);
                clueAutoResult.setFilterId(ksFlowGraphId);
                clueAutoResult.setFilterName(filterName);
                clueAutoResult.setDbCode(item.getAs("DB_CODE"));
                clueAutoResult.setTableName(item.getAs("TABLE_NAME"));
                String id = "";
                if (!Objects.isNull(item.getAs("id"))) {
                    id = item.getAs("id").toString();
                }
                clueAutoResult.setSourceId(id);
                String msg = "";
                if (!Objects.isNull(item.getAs("msg"))) {
                    msg = item.getAs("msg").toString();
                }
                clueAutoResult.setReason(msg);
                clueAutoResult.setTips(msg);
                clueAutoResult.setPower(RuleUtil.powerConvert(power, item));
                clueAutoResult.setResultPower(RuleUtil.powerConvert(power, item));
                StaticCollection.getClueAutoResultList(ksFlowGraphId).add(clueAutoResult);
            });


            KsFilterStatistics ksFilterStatistics = queryFactory.selectFrom(qKsFilterStatistics).where(qKsFilterStatistics.ruleId.eq(finalRuleId)).orderBy(qKsFilterStatistics.id.desc()).fetchFirst();
            ksFilterStatistics.setClueResultTotal(ksFilterStatistics.getClueResultTotal() + StaticCollection.getClueAutoResultList(ksFlowGraphId).size());
            ksFilterStatisticsService.save(ksFilterStatistics);

            clueAutoResultService.saveAllPartition(StaticCollection.getClueAutoResultList(ksFlowGraphId), 1000, processParam.getKsFlowGraph().getAuth());
            StaticCollection.removeClueAutoResultList(ksFlowGraphId);
        }
        return SparkResult.success(ruleResultDataset);
    }

    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.KS_RULE;
    }
}
